package com.zwh.se.collection.concurrent.blocking_queue;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @author yexi
 * <p>
 * 阻塞队列(BlockingQueue)：首先它是一个队列(Queue),而队列继承于Collection,也就是说它是一个具有阻塞特点的队列。
 * 什么是阻塞：
 * 非阻塞:入队：当队列满时放入元素,直接丢失
 * 出队：当队列没有元素时,取出null
 * 阻塞:入队：当队列满时放入元素,等待(阻塞),直至队列中有元素出队,然后再放进去
 * 出队：当队列没有元素时,等待,直至队列中有元素入队,然后才取出来
 * <p>
 * 概括：首先它是一个队列,其次它还支持在检索元素时,等待直至队列变为非空,以及在存储元素时,等待直至队列中的空间变为可用的操作。
 * <p>
 * 一些特点：
 * 阻塞队列不接受null元素。在尝试add 、 put或offer null时，实现会抛出NullPointerException。null用作标记值以指示poll操作失败。
 * BlockingQueue方法有四种形式，以不同的方式处理不能立即满足但可能在未来某个时刻满足的操作：
 * (1):抛出异常：插入add(e)、删除remove()、检索element()
 * (2):返回操作结果(true和false)：插入offer(e)、删除poll()、检索peek()
 * (3):阻塞：插入put(e)、删除take()
 * (4):超时阻塞：插入offer(e, time, unit)、删除poll(time, unit)
 * <p>
 * 本demo中为通过Array(数组)所实现的阻塞队列。
 * ArrayBlockingQueue：一个底层基于数组的有边界的(有界限的)阻塞队列。此队列对元素进行 FIFO(先进先出)排序。
 * 队列的头部是在队列中放置时间最长的元素。队列的尾部是在队列中放置时间最短的元素。新元素会被插入到队列的尾部，队列检索操作会获取队列头部的元素。
 * ArrayBlockingQueue 不支持读写同时操作，底层基于数组。
 */
@SuppressWarnings({"rawtypes", "unchecked"})
public class ArrayBlockingQueueDemo {
    public static void main(String[] args) throws InterruptedException {
        //创建一个队列，队列可以指定容量指定长度3：
        ArrayBlockingQueue arrayBlockingQueue = new ArrayBlockingQueue(3);
        //创建一个空队列(没有元素,方便后面写demo)
        ArrayBlockingQueue noneBlockingQueue = new ArrayBlockingQueue<>(1);

        System.out.println("--------------------[1]新增操作--------------------");
        //阻塞队列不可以添加null元素，会报空指针异常：NullPointerException，只在该类演示一次。
        try {
            arrayBlockingQueue.add(null);
        } catch (Exception e) {
            System.out.println("\033[31;4m" + e + "\033[0m");
        }
        try {
            arrayBlockingQueue.offer(null);
        } catch (Exception e) {
            System.out.println("\033[31;4m" + e + "\033[0m");
        }
        try {
            arrayBlockingQueue.put(null);
        } catch (Exception e) {
            System.out.println("\033[31;4m" + e + "\033[0m");
        }
        //正常添加
        arrayBlockingQueue.add("aaa");
        arrayBlockingQueue.offer("bbb");
        arrayBlockingQueue.put("ccc");
        //[aaa, bbb, ccc]
        System.out.println(arrayBlockingQueue);
        //在队列满的情况下,再添加元素：
        //1、报错
        try {
            arrayBlockingQueue.add("ddd");
        } catch (Exception e) {
            System.out.println("\033[31;4m" + e + "\033[0m");
        }
        //2、返回操作结果
        System.out.println(arrayBlockingQueue.offer("ddd"));
        //3、阻塞(直至有其他线程将阻塞队列里的元素出队)
        Thread thread = new Thread(() -> {
            try {
                //该段代码模拟其他线程各类操作所消耗的时长
                Thread.sleep(2000);
                //3秒后从阻塞队列中出队一个元素
                arrayBlockingQueue.poll();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
        thread.start();
        //直至上方的子线程将元素出队前,将一直处于堵塞状态
        arrayBlockingQueue.put("ddd");
        System.out.println(arrayBlockingQueue);
        //4、设置最大阻塞时间，如果时间到了，队列还是满的，就不再阻塞了
        System.out.println(arrayBlockingQueue.offer("ddd", 2, TimeUnit.SECONDS));


        System.out.println("--------------------[2]检索操作--------------------");
        System.out.println(arrayBlockingQueue.peek());
        System.out.println(arrayBlockingQueue);
        System.out.println(arrayBlockingQueue.element());
        System.out.println(arrayBlockingQueue);

        //获取不到时只返回操作结果
        System.out.println(noneBlockingQueue.peek());
        //获取不到时报错
        try {
            System.out.println(noneBlockingQueue.element());
        } catch (Exception e) {
            System.out.println("\033[31;4m" + e + "\033[0m");
        }

        System.out.println("--------------------[3]删除操作(获取头元素并使其出队)--------------------");
        System.out.println(arrayBlockingQueue.poll());
        System.out.println(arrayBlockingQueue);
        System.out.println(arrayBlockingQueue.take());
        System.out.println(arrayBlockingQueue);
        System.out.println(arrayBlockingQueue.remove());
        System.out.println(arrayBlockingQueue);

        //在队列空的情况下,再出队：
        //1、报错
        try {
            System.out.println(noneBlockingQueue.remove());
        } catch (Exception e) {
            System.out.println("\033[31;4m" + e + "\033[0m");
        }
        //2、返回获取到的元素(空队列的话返回null)
        System.out.println(noneBlockingQueue.poll());
        //3、设置最大阻塞时间，如果时间到了，队列还是空的，就不再阻塞了，并返回null
        System.out.println(noneBlockingQueue.poll(2, TimeUnit.SECONDS));
        //4、阻塞(队列为空，永远阻塞)
        Thread thread2 = new Thread(() -> {
            try {
                //该段代码模拟其他线程各类操作所消耗的时长
                Thread.sleep(2000);
                noneBlockingQueue.put("eee");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
        thread2.start();
        System.out.println(noneBlockingQueue.take());

        System.out.println("--------------------[4]源码解析--------------------");
        /**
         * @see ArrayBlockingQueue#items 底层存储的Object数组
         * @see ArrayBlockingQueue#takeIndex 取元素用到的索引，初始结果为0
         * @see ArrayBlockingQueue#putIndex 放元素用到的索引，初始结果为0
         * @see ArrayBlockingQueue#count 数组中的元素个数
         * @see ArrayBlockingQueue#lock ReentrantLock。定义为属性必然是因为很多方法都用到了。
         * @see ArrayBlockingQueue#notEmpty  锁伴随的一个等待队列
         * @see ArrayBlockingQueue#notFull   锁伴随的一个等待队列
         *
         * // capacity：指定阻塞队列固定容量
         * @see ArrayBlockingQueue#ArrayBlockingQueue(int capacity)
         *         this(capacity, false);
         * // fair：如果为true,则在插入或删除时线程对阻塞队列的访问将按 FIFO 顺序处理;如果为false,则未指定访问顺序。
         * @see ArrayBlockingQueue#ArrayBlockingQueue(int capacity, boolean fair)
         *         if (capacity <= 0)    //健壮性考虑
         *             throw new IllegalArgumentException();
         *         this.items = new Object[capacity];  //初始化底层数组
         *         lock = new ReentrantLock(fair);   //初始化锁
         *         notEmpty = lock.newCondition();   //初始化等待队列
         *         notFull =  lock.newCondition();   //初始化等待队列
         */
        ArrayBlockingQueue testBlockingQueue = new ArrayBlockingQueue<>(1);
        /**
         * //入队的方法,是其他方法的基础。仅在持有锁时调用。
         * @see ArrayBlockingQueue#enqueue(Object x)
         *         // assert lock.getHoldCount() == 1;
         *         // assert items[putIndex] == null;
         *         final Object[] items = this.items;  // 底层数组赋给items
         *         items[putIndex] = x;  // 在对应的下标位置放入元素
         *         if (++putIndex == items.length) // 如果放入 putIndex+1后的结果 == 数组容量
         *             putIndex = 0;   // 则将putIndex置为0
         *         count++;  // 已存入元素数量+1
         *         notEmpty.signal();
         *
         * @see ArrayBlockingQueue#put(Object e)
         *         checkNotNull(e);
         *         final ReentrantLock lock = this.lock;
         *         lock.lockInterruptibly();
         *         try {
         *             //这里为什么不能使用if?是为了避免虚假唤醒。
         *             while (count == items.length)   //当队列放满的时候
         *                 notFull.await();
         *             enqueue(e);
         *         } finally {
         *             lock.unlock();
         *         }
         */
        testBlockingQueue.put("test");
        /**
         * //出队的方法,是其他方法的基础。仅在持有锁时调用。
         * @see ArrayBlockingQueue#dequeue()
         *         // assert lock.getHoldCount() == 1;
         *         // assert items[takeIndex] != null;
         *         final Object[] items = this.items;  //底层数组赋给items
         *         @SuppressWarnings("unchecked")
         *         E x = (E) items[takeIndex];  //在对应的位置取出元素
         *         items[takeIndex] = null;  //对应位置元素取出后就置为null
         *         if (++takeIndex == items.length)  //++takeIndex 加1操作
         *             takeIndex = 0; (先进先出的真正实现,由于堵塞,如果取元素取到最后一位,则后进的就必然是0开始)
         *         count--;  //每取出一个元素，count减1操作
         *         if (itrs != null)
         *             itrs.elementDequeued();
         *         notFull.signal();
         *         return x;  //将取出的元素作为方法的返回值
         */
        testBlockingQueue.take();


    }
}
